# RocketMQ 快速启动

## 1. 概述

使用 Docker-compose 快速部署一个单节点单副本 RocketMQ 服务，并完成简单的消息收发。

> 参考文档 https://rocketmq.apache.org/zh/docs/quickStart/03quickstartWithDockercompose

## 2. 运行准备

### 下载示例程序

```powershell
# Bash/Shell

# 下载示例程序
git clone https://gitee.com/leo-blaze/examples.git
# 启动一个单节点单副本 RocketMQ
cd examples/database/rocketmq-quickstart/script
# windows
docker-compose -p rockermq_project up -d
```

### 启动一个单节点单副本 RocketMQ

```powershell
# Bash/Shell
cd examples/database/rocketmq-quickstart/script
# windows
docker-compose -p rockermq_project up -d
```

### 清除一个单节点单副本 RocketMQ（测试完删除命令）

```powershell
# Bash/Shell
cd examples/database/rocketmq-quickstart/script
# windows  停止但不删除容器
docker-compose -p rockermq_project stop
# 停止并删除容器
docker-compose -p rockermq_project down
# 停止并删除容器和网络
docker-compose -p rockermq_project down --remove-orphans
# 停止并删除容器、网络、镜像和数据卷
docker-compose -p rockermq_project down -v --rmi all
```

## 3. 运行示例程序

### 创建一个 Topic 用于测试

```powershell
$ docker exec -it rmqbroker bash
$ sh mqadmin updatetopic -t TestTopic -c DefaultCluster
```

### 向 TestTopic 内发送消息示例

```java
# examples/database/rocketmq-quickstart/src/main/java/rocketmq/quickstart/ProducerExample.java
import java.io.IOException;

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ProducerExample {
  private static final Logger logger = LoggerFactory.getLogger(ProducerExample.class);

  public static void main(String[] args) throws ClientException, IOException {
    // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8080;xxx:8081
    String endpoint = "localhost:8081";
    // 消息发送的目标Topic名称，需要提前创建。
    String topic = "TestTopic";
    ClientServiceProvider provider = ClientServiceProvider.loadService();
    ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
    ClientConfiguration configuration = builder.build();
    // 初始化Producer时需要设置通信配置以及预绑定的Topic。
    Producer producer = provider.newProducerBuilder()
        .setTopics(topic)
        .setClientConfiguration(configuration)
        .build();
    // 普通消息发送。
    Message message = provider.newMessageBuilder()
        .setTopic(topic)
        // 设置消息索引键，可根据关键字精确查找某条消息。
        .setKeys("messageKey")
        // 设置消息Tag，用于消费端根据指定Tag过滤消息。
        .setTag("messageTag")
        // 消息体。
        .setBody("messageBody".getBytes())
        .build();
    try {
      // 发送消息，需要关注发送结果，并捕获失败等异常。
      SendReceipt sendReceipt = producer.send(message);
      logger.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
    } catch (ClientException e) {
      logger.error("Failed to send message", e);
    }
    producer.close();
  }
}
```

### 从 TestTopic 接收消息示例

```java
import java.io.IOException;
import java.util.Collections;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PushConsumerExample {
  private static final Logger logger = LoggerFactory.getLogger(PushConsumerExample.class);

  private PushConsumerExample() {
  }

  public static void main(String[] args) throws ClientException, IOException, InterruptedException {
    final ClientServiceProvider provider = ClientServiceProvider.loadService();
    // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8080;xxx:8081
    String endpoints = "localhost:8081";
    ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
        .setEndpoints(endpoints)
        .build();
    // 订阅消息的过滤规则，表示订阅所有Tag的消息。
    String tag = "*";
    FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
    // 为消费者指定所属的消费者分组，Group需要提前创建。
    String consumerGroup = "YourConsumerGroup";
    // 指定需要订阅哪个目标Topic，Topic需要提前创建。
    String topic = "TestTopic";
    // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
    PushConsumer pushConsumer = provider.newPushConsumerBuilder()
        .setClientConfiguration(clientConfiguration)
        // 设置消费者分组。
        .setConsumerGroup(consumerGroup)
        // 设置预绑定的订阅关系。
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        // 设置消费监听器。
        .setMessageListener(messageView -> {
          // 处理消息并返回消费结果。
          logger.info("Consume message successfully, messageId={}", messageView.getMessageId());
          return ConsumeResult.SUCCESS;
        })
        .build();
    Thread.sleep(Long.MAX_VALUE);
    // 如果不需要再使用 PushConsumer，可关闭该实例。
    // pushConsumer.close();
  }
}
```
